package org.ricks.gateway.cluster;

import io.scalecube.cluster.Cluster;
import io.scalecube.cluster.ClusterMessageHandler;
import io.scalecube.cluster.Member;
import io.scalecube.cluster.membership.MembershipEvent;
import io.scalecube.cluster.transport.api.Message;
import org.ricks.ioc.utils.Logger;
import reactor.core.publisher.Flux;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Function;

// 基于 GossipProtocolImpl 扩展事件传播[11](@ref)
public class EnhancedGossipProtocol implements ClusterMessageHandler, Function<Cluster, ClusterMessageHandler> {
    private Cluster cluster;
    private final Map<String, MemberState> memberStates = new ConcurrentHashMap<>();


    private final MetadataResolver metadataResolver;

    public EnhancedGossipProtocol(MetadataResolver metadataResolver) {
        this.metadataResolver = metadataResolver;
    }

    private static class MemberState {
        long lastSeen;
        boolean suspect;
    }


    @Override
    public void onMessage(Message message) {
//        cluster.spreadGossip(message);
//        cluster.updateMetadata(message);
        // 处理自定义消息
        String senderId = message.sender();
        updateMemberState(senderId);
    }

    @Override
    public ClusterMessageHandler apply(Cluster cluster) {
        this.cluster = cluster;
        startMembershipMonitoring();
        return this;
    }

    @Override
    public void onMembershipEvent(MembershipEvent event) {
        Member member = event.member();

        switch (event.type()) {
            case ADDED -> onJoin(member);
            case LEAVING -> onSuspect(member);
            case REMOVED -> onLeave(member);
        }
    }


    public void onJoin(Member member) {
        try {
            Optional<Broker> brokerOpt = metadataResolver.resolveBroker();
            if (brokerOpt.isPresent()) {
                Broker broker = brokerOpt.get();
                broker.setSuspect(false); // 重置可疑状态
                LoadBalancer.addBroker(broker);
                Logger.info("Node joined: {}", broker.getClusterAddress());
            } else {
                Logger.warn("No broker metadata found for member: {}", member.id());
            }
        } catch (Exception e) {
            Logger.error("Error processing join event for member: {}", e, member.id());
        }
    }

    public void onSuspect(Member member) {
        try {
            Optional<Broker> brokerOpt = metadataResolver.resolveBroker();
            if (brokerOpt.isPresent()) {
                Broker broker = brokerOpt.get();
                LoadBalancer.markSuspect(broker);
                Logger.warn("Node suspected: {}", broker.getClusterAddress());
            }
        } catch (Exception e) {
            Logger.error("Error processing suspect event for member: {}", e, member.id());
        }
    }

    public void onLeave(Member member) {
        try {
            Optional<Broker> brokerOpt = metadataResolver.resolveBroker();
            if (brokerOpt.isPresent()) {
                Broker broker = brokerOpt.get();
                LoadBalancer.removeBroker(broker);
                Logger.warn("Node left: {}", broker.getClusterAddress());
            }
        } catch (Exception e) {
            Logger.error("Error processing leave event for member: {}",e,  member.id());
        }
    }


    private void startMembershipMonitoring() {
        // 定期检查成员状态
        Flux.interval(Duration.ofSeconds(2)).subscribe(_ -> checkMemberStates());

        // 初始添加所有成员
        cluster.members().forEach(member -> {
            memberStates.put(member.id(), new MemberState());
        });
    }

    private void updateMemberState(String memberId) {
        MemberState state = memberStates.computeIfAbsent(memberId, k -> new MemberState());
        state.lastSeen = System.currentTimeMillis();
        state.suspect = false;
    }

    private void checkMemberStates() {
        long now = System.currentTimeMillis();
        Set<String> currentMemberIds = new HashSet<>();

        // 检查所有已知成员
        for (Member member : cluster.members()) {
            currentMemberIds.add(member.id());
            MemberState state = memberStates.get(member.id());

            if (state == null) {
                // 新成员加入
                state = new MemberState();
                state.lastSeen = now;
                memberStates.put(member.id(), state);
            } else if (now - state.lastSeen > 5000 && !state.suspect) {
                // 超过5秒未更新，标记为可疑
                state.suspect = true;
            }
        }

        // 检查已离开的成员
        Iterator<Map.Entry<String, MemberState>> it = memberStates.entrySet().iterator();
        while (it.hasNext()) {
            Map.Entry<String, MemberState> entry = it.next();
            String memberId = entry.getKey();

            if (!currentMemberIds.contains(memberId)) {
                // 成员已离开
                it.remove();
                Optional<Member> member = cluster.memberById(memberId);
            }
        }
    }
}
